/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
/**
 * Software License, Version 1.0
 * 
 * Copyright 2004 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.gridapps.nbgridftp.publisher;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Properties;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.jms.NBJmsInitializer;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceBulletin;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceNotification;
import cgl.narada.service.Task;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.NBRecoveryListener;
import cgl.narada.service.client.NBRecoveryNotification;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ProducerConstraints;
/**
   @author Sang Boem Lim
   $Date$
   $Revision$
 */

class Client2NBControl extends Thread implements javax.jms.MessageListener,
    NBEventListener, NBRecoveryListener, ServiceBulletin{

    private  InputStream  fromClient;
    private  OutputStream toClient;

    private int port;
    private  Client2NBData data;
    
    private  final String WELCOME = "WELCOME: ";
    private  final String INTER   = "INTER: ";
    private  final String CONTROL = "CONTROL: ";

    private  int jobID = 0;
    private  int oldJobID = 0;

    // Naradabrokering...
    private ClientService clientService;
    private String moduleName= "FragmentationClient: ";
    private EventConsumer consumer;
    private  EventProducer producer;
    private  ProducerConstraints producerConstraints, producerConstraints2;

    private String tmpdir;
    public Client2NBControl (int port, Properties connectionProperties,
                             String tmpdir) {

        this.port = port;

        this.tmpdir = tmpdir;
        data = new Client2NBData(this, tmpdir);

        // NB set up..
        String transportType = "niotcp";
        int entityId = 7007;
        int templateId = 12345;

        try {
            initializeChatSession((TopicConnectionFactory) 
                                  new NBJmsInitializer(connectionProperties, 
                                                        transportType, entityId).lookup());

            clientService = SessionService.getClientService(entityId, this);
            clientService.initializeBrokerCommunications(connectionProperties,
                                                         transportType);

            initializeProducerAndConsumer(templateId);

        } catch (Exception e){
            System.err.println(e);
            System.exit(1);
        }
    }

    public void run() {

        try{
            ServerSocket server = new ServerSocket(port);
            while(true) {
               
                while(!isFinished()) {
                    try{
                        Thread.sleep(1000);
                    } catch (InterruptedException err){}

                }

                data.delete();

                Socket s = server.accept();
                isFirstTime = true;

                isClosed = false;
                jobID++;
                
                data = new Client2NBData(this, tmpdir);

                fromClient = s.getInputStream();
                toClient   = s.getOutputStream();
                
                clientName = s.getInetAddress().getHostAddress() + ":" + 
                    s.getPort();

                readFromClient();

                if (toClient != null) toClient.close();
                if (fromClient != null) fromClient.close();
                if (s != null) s.close();

                data.close();

                welcome = null;                
            }
        } catch (Exception e){
            System.exit(1);
        }
    }

    private synchronized  boolean isFinished() {

        return (jobID == oldJobID);
    }

    //********* NB Start *************************************8
    private  TopicSession pubSession;
    private TopicSession subSession;
    private  TopicPublisher publisher;
    private TopicConnection connection;
    private  String clientName;

    private String welcome = null;
    private String hostPort;

    private  boolean isClosed = false;
    /**
     * Receive message from topic subscriber. This will only be subscribed
     * a message issued to "/slim/GridFTP/server".
     * 
     */
    public void onMessage(Message message) {

        try {
            if (message.getJMSType().equals("BytesMessage")) {

                BytesMessage byteMessage = (BytesMessage) message;
                byte[] byteHeader = new byte[byteMessage.readInt()];

                byteMessage.readBytes(byteHeader);

                String header = new String(byteHeader);
                if(header.startsWith("From server control:")){
                    int length = byteMessage.readInt();
                    byte[] reply = new byte[length];
                    
                    byteMessage.readBytes(reply);
                    
                    if (toClient != null) {
                        toClient.write(reply, 0, length);
                        toClient.flush();
                    } else{
                        if(!isClosed) runCleanup();
                    }
                } else if (header.startsWith("From server data:")) {
                    
                    int length = byteMessage.readInt();
                    byte[] reply = new byte[length];
                    
                    byteMessage.readBytes(reply);
                    data.write2Client(reply, length);
                } else if (header.startsWith("SERVER CLOSED:")) {

                    if (!isClosed) runCleanup();
                }
            } else if (message.getJMSType().equals("TextMessage")){

                welcome = ((TextMessage) message).getText();

                int idx = welcome.indexOf(":::");
                hostPort = welcome.substring(idx + 3);
                welcome = welcome.substring(0, idx);
            }
        } catch(JMSException jmse){ jmse.printStackTrace(); 
        } catch(IOException e){

            if (!isClosed) runCleanup();
        }
    }

    /* Create and send message using topic publisher */
    public  void writeMessage(byte[] buf, int length, String str) 
        throws JMSException {

        byte[] header = (str + clientName).getBytes();

        BytesMessage byteMessage = pubSession.createBytesMessage();
        byteMessage.writeInt(header.length);
        byteMessage.writeBytes(header);
        byteMessage.writeInt(length);
        byteMessage.writeBytes(buf);
        
        publisher.publish(byteMessage); 
    }

    private void startConnection() throws JMSException {
        connection.start();
    }
  
    private void stopConnection() throws JMSException {
        connection.stop();
    }

    /* Close the JMS connection */
    private void close() throws JMSException {
        connection.close();
    }

    /* Initializes the instance variables */
    private void set(TopicConnection con, TopicSession pubSess,
                    TopicSession subSess, TopicPublisher pub) {

        this.connection = con;
        this.pubSession = pubSess;
        this.subSession = subSess;
        this.publisher = pub;
    }
  
    private void initializeChatSession(TopicConnectionFactory conFactory) 
        throws Exception {

        // Create a JMS connection
        TopicConnection connection = 
            conFactory.createTopicConnection("guest","password");
    
        // Create a JMS session object
        TopicSession pubSession =
            connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

        TopicSession subSession =
            connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    
        String serverString = "/slim/GridFTP/server";
        String clientString = "/slim/GridFTP/client";

        // Lookup a JMS topic
        Topic serverTopic = pubSession.createTopic(serverString);  
        Topic clientTopic = pubSession.createTopic(clientString);  
    
        // Create a JMS publisher and subscriber
        TopicPublisher publisher = pubSession.createPublisher(clientTopic);
        TopicSubscriber subscriber = subSession.createSubscriber(serverTopic);
    
        // Set a JMS message listener
        subscriber.setMessageListener(this);
    
        // Intialize the Chat application
        set(connection, pubSession, subSession, publisher);
    
        // Start the JMS connection; allows messages to be delivered
        connection.start();
    }

    // naradabrokering
    public void initializeProducerAndConsumer(int templateId) 
        throws ServiceException {

        Profile profile = 
            clientService.createProfile(TemplateProfileAndSynopsisTypes.STRING, 
                                        "Movies/Casablanca");
        consumer = clientService.createEventConsumer(this);
        
        //ConsumerConstraints constraints = 
        //  consumer.createConsumerConstraints(profile);
        
        //consumer.subscribeTo(profile);
        
        producer = clientService.createEventProducer();
        producer.setSuppressRedistributionToSource(true);
        TemplateInfo templateInfo = 
            clientService.createTemplateInfo(templateId, 
                                             TemplateProfileAndSynopsisTypes.STRING,
                                             "Movies/Casablanca");
        producerConstraints = 
            producer.createProducerConstraints(templateInfo);
        producerConstraints2 = 
            producer.createProducerConstraints(templateInfo);
        long recoveryId = producer.recover(templateId, this);
        System.out.println(moduleName + "Assigned recovery id = [" +
                           recoveryId + "] \n\n");
        
    }
    
    
    public  void publishEvent(String stringMsg,
                             Properties fragmentationProperties) 
        throws ServiceException {
        if (stringMsg.equals("")) {
            stringMsg += System.currentTimeMillis();
        }
        producer.generateEventIdentifier(true);
        producer.setTemplateId(12345);
        producer.setDisableTimestamp(false);
        NBEvent nbEvent = 
            producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                   "Movies/Casablanca",
                                   stringMsg.getBytes());
        
        producerConstraints.setSendAfterFragmentation(fragmentationProperties);
        
        producer.publishEvent(nbEvent, producerConstraints);
    }
    
    public void publishEvent(String stringMsg) throws ServiceException {
        if (stringMsg.equals("")) {
            stringMsg += System.currentTimeMillis();
        }
        producer.generateEventIdentifier(true);
        producer.setTemplateId(12345);
        producer.setDisableTimestamp(false);
        NBEvent nbEvent = 
            producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                   "Movies/Casablanca",
                                   stringMsg.getBytes());
        producer.publishEvent(nbEvent, producerConstraints2);
    }
    
    
    public void onEvent(NBEvent nbEvent) {
        System.out.println("\n\n\n" + moduleName + "Received COALESCED file [" + 
                           new String( nbEvent.getContentPayload() )   +
                           "] of length=" + 
                           nbEvent.getEventHeaders().getOriginalPayloadLength());
        if (progress != null) progress.setDone();
    }
    
    
    /** Upon completion of the attempt to recover, this method is invoked on
     the listener that was registered with the */
    public void onRecovery(NBRecoveryNotification recoveryNotification) {
        System.out.println(moduleName + recoveryNotification);
    }
    
    
    public void onServiceException(ServiceException serviceException) {
        System.out.println(moduleName + serviceException);
    }
    
    
    private boolean isFirstTime = true;
    private  NBProgress progress;
    public void onServiceNotification(ServiceNotification serviceNotification){

        if (serviceNotification.hasTaskInfo()) {
            Task task = serviceNotification.getTaskInfo();

            boolean isSubtasks = task.hasSubTasks();

            int subTasks = 1;
            if(isSubtasks)
                subTasks = task.getNumOfSubTasks();

            if(task.getStatus() == Task.COMPLETED) {
                if(progress != null) progress.setDone();
            } else {
                if(isFirstTime) {
                    isFirstTime = false;
                    progress = new NBProgress(subTasks, data.getFileName());
                    
                    if(isSubtasks) {
                        for(int i = 0; i < subTasks; i++)
                            progress.setup(i, task.getSubTask(i).getNumOfWorkUnits());
                    } else 
                        progress.setup(0, task.getNumOfWorkUnits());

                    progress.start();
                }

                if(isSubtasks) {
                    for(int i = 0; i < subTasks; i++)
                        progress.update(i, task.getSubTask(i).getNumOfWorkUnitsCompleted());
                } else 
                    progress.setup(0, task.getNumOfWorkUnitsCompleted());
            }
            //System.out.println(task.toString());
        }
    }
    
    //********* NB end *************************************8
    
    private String readLine() throws IOException {
        int c = fromClient.read();
        if (c == -1) {
            return null;
        }

        StringBuffer buf = new StringBuffer();
        buf.append((char)c);
        while((c = fromClient.read()) != -1 ) {
            if (c == '\r') {
                c = fromClient.read();
                if (c == '\n' || c == -1) {
                    break;
                } else {
                    throw new IOException("bad format");
                }
            } else {
                buf.append( (char)c);
            }
        }

        return buf.toString();
    }

    private  void runCleanup() {

        isClosed = true;

        while((data != null) && !data.isFinished()) {
            try{
                Thread.sleep(1000);
            } catch (InterruptedException err){}
        }
        
        try {
            writeMessage(new byte[0], 0, INTER);
            
            if (toClient != null) toClient.close();
            if (fromClient != null) fromClient.close();
        } catch (IOException err) {
        } catch (JMSException err){}

        if(progress.isAlive()) {
            progress.setDone();
        }
        oldJobID++;
    }

    public  void dataClosed() {

        if (!isClosed) runCleanup();
    }

    /**
     * This method is connecting to a client and download datas into the local 
     * temporarily file.
     **/
    private void readFromClient(){

        try {
            
            // I have to send this information whenever a client connect
            // to this proxy to simulate FTP server.
            if (welcome != null) 
                toClient.write(welcome.getBytes());
            else 
                toClient.write("220 atlas.ucs.indiana.edu FTP server (GridFTP Server 1.5 [GSI patch v0.5] wu-2.6.2(2) Thu Jun 19 14:48:58 CDT 2003) ready.\n".getBytes());
            toClient.flush();

            byte [] inToken  = new byte[1024];
            byte [] outToken = null;
            
            int bytesRead;
            boolean normal = false;

            while((bytesRead = fromClient.read(inToken)) != -1) {
                String str = new String(inToken, 0, bytesRead);

                if(str.startsWith("NB")) {

                    String msg = readLine();
                    
                    if(msg.startsWith("STOR::")) {
                        data.setFileName(msg.substring(6));
                        data.start();
                    } else {
                        toClient.write((hostPort + "\r\n").getBytes());
                        toClient.flush();
                        
                        data.setHostPort(SendProxy.getHost(msg),
                                         Integer.parseInt(SendProxy.getPort(msg)));
                        
                    }
                } else if(str.startsWith("FINISH")) {
                    normal = true;
                }else {
                    if (welcome != null)
                        writeMessage(inToken, bytesRead, CONTROL);
                    else
                        writeMessage(inToken, bytesRead, WELCOME);
                }
            }
            
            if (normal) {
                oldJobID++;
                isClosed = true;
                writeMessage(new byte[0], 0, "NORMAL: ");
            } else {
                writeMessage(new byte[0], 0, "ABNORMAL: ");
            }
        } catch (Exception e) {
            if (!isClosed) runCleanup();
        }
    }
}
